package com.ydl.learning.flink.demo.udf

import com.ydl.learning.flink.demo.TableUtils
import org.apache.flink.table.api._
import org.apache.flink.table.functions.ScalarFunction

/**
 *
 *
 * @author ydl
 * @since 2022/3/11
 */
object UdfDemo extends TableUtils {

  // 自定义udf
  class SubstringFunction extends ScalarFunction {
    def eval(s: String, begin: Integer, end: Integer): String = {
      s.substring(begin, end)
    }
  }

  def useUdf(): Unit = {
    val env = tableEnv

// 在 Table API 里不经注册直接“内联”调用函数
    env.from("MyTable").select(call(classOf[SubstringFunction], $"myField", 5, 12))

// 注册函数
    env.createTemporarySystemFunction("SubstringFunction", classOf[SubstringFunction])

// 在 Table API 里调用注册好的函数
    env.from("MyTable").select(call("SubstringFunction", $"myField", 5, 12))

// 在 SQL 里调用注册好的函数
    env.sqlQuery("SELECT SubstringFunction(myField, 5, 12) FROM MyTable")

  }

}
